package ccm.communicate.smctp;

import ccm.common.*;
import ccm.communicate.common.Communicate;
import ccm.communicate.common.NotStartException;
import ccm.communicate.common.SendFailedException;

import javax.swing.*;
import java.util.Arrays;
import java.util.Calendar;
import java.util.LinkedList;
import java.util.TimeZone;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import static ccm.common.Utils.timeFormatter;

/**
 * <p>
 * SimpleMultiChannelTransportProtocol <br/>
 * 简易多通道传输协议 <br/>
 * </p>
 * <p>
 * 长度    2/4字节小段序                 <br/>
 * 时间戳  4/8字节小段序                 <br/>
 * 通道    1字节                        <br/>
 * 版本    1字节                        低6位：版本，第6位 短时间戳，第7位短长度<br/>
 * 校验    2字节小段序                   <br/>
 * 包头    3字节       0XFF 0X12 0X17   <br/>
 * 数据 <br/>
 * </p>
 *
 * <p>
 * 发送条件 不在操作&&(100ms未更新||传输效率90%+||达到缓冲区半满) <br/>
 * len>=包头/(1-传输效率) <br/>
 * </p>
 * <p>
 * 把当前通道的ID转移进正在发送区,重新分配缓冲区
 * </p>
 */
public class Smctp extends Communicate implements StatusChangeListener
{
    public final static int CHANNEL_DEBUG=0;
    public final static int CHANNEL_MENU=1;
    public final static int CHANNEL_VISUAL_SCOPE=2;
    protected final static int CHANNEL_MAX=3;
    protected static final int smctpVersion=1;
    protected static final int zoneOffset=Calendar.getInstance(TimeZone.getDefault()).get(Calendar.ZONE_OFFSET);

    static
    {
        CCMConfigManager.setIntegerIfAbsent("ccm/communicate/smctp/Smctp/designatedDataChannel",CHANNEL_MAX);
    }

    protected final ReceiveInfo RVI;
    protected final LinkedList<Pair<Long,Long>> inSpeedHistory;
    protected final LinkedList<Pair<Long,Long>> outSpeedHistory;
    protected final LinkedList<ReceiveDataEvent>[] receiveDataList;
    protected final LinkedList<byte[]>[] sendDataList;
    protected long inSpeedSum;
    protected long outSpeedSum;
    protected boolean start;
    /**
     * 上一次接收的US时间戳
     */
    protected long lastReceiveUS;
    protected volatile boolean sentFlag;
    protected int successBytes;
    protected int totalBytes;
    /**
     * 指定的的分发通道
     */
    protected int designatedDataChannel=CCMConfigManager.getInteger("ccm/communicate/smctp/Smctp/designatedDataChannel");
    private int currentConnectId=0;
    private long currentConnectLastUs=0;
    private String currentConnectName=null;
    private Communicate next;

    public Smctp()
    {
        super(CHANNEL_MAX);
        RVI=new ReceiveInfo();
        inSpeedHistory=new LinkedList<>();
        outSpeedHistory=new LinkedList<>();
        receiveDataList=new LinkedList[CHANNEL_MAX];
        sendDataList=new LinkedList[CHANNEL_MAX];
        start=false;
        for(int i=0;i<CHANNEL_MAX;++i)
        {
            receiveDataList[i]=new LinkedList<>();
            sendDataList[i]=new LinkedList<>();
            new Thread(new Consumer(i)).start();
        }
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(new Sender(),0,4,TimeUnit.MILLISECONDS);
        clear();
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(()->
        {
            try
            {
                if(currentConnectName==null)
                {
                    sendInfoRequest();
                }
                else
                {
                    sendHeartBeat();
                }
            }catch(NotStartException|SendFailedException ignored)
            {

            }catch(Exception e)
            {
                e.printStackTrace();
            }
        },0,2,TimeUnit.SECONDS);
        Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(()->
        {
            try
            {
                if(currentConnectName!=null)
                {
                    sendInfoRequest();
                }
            }catch(NotStartException|SendFailedException ignored)
            {

            }catch(Exception e)
            {
                e.printStackTrace();
            }

        },0,20,TimeUnit.SECONDS);
    }

    protected void sendInfoRequest()
    {
        if(!isOpen())
        {
            return;
        }
        System.out.println("SMCTP send system info request");
        byte[] tmp=new byte[8];
        Buffer.setUint64(tmp,0,currentConnectLastUs);
        send(tmp,0XFF,0X19);
    }

    protected void sendHeartBeat()
    {
        if(!isOpen())
        {
            return;
        }
        if(!sentFlag)
        {
            sentFlag=false;
            send(new byte[0],0XFF);
            System.out.println("SMCTP send HeartBeat");
        }
    }

    public int getSuccessBytes()
    {
        return successBytes;
    }

    public int getTotalBytes()
    {
        return totalBytes;
    }

    public String getCurrentConnect()
    {
        if(currentConnectName==null)
        {
            return "";
        }
        return currentConnectId+","+currentConnectName;
    }

    public void setNext(Communicate nextIn,int nextChannel)
    {
        if(next!=null)
        {
            next.stop();
        }
        stop();
        next=nextIn;
        next.addStatusChangeListener(this);
        next.addReceiveDataListener(this,nextChannel);
    }

    /**
     * 清空
     */
    protected void clear()
    {
        RVI.clear();
        lastReceiveUS=0;
        inSpeedHistory.clear();
        outSpeedHistory.clear();
        inSpeedSum=0;
        outSpeedSum=0;
        successBytes=0;
        totalBytes=0;

        currentConnectId=0;
        currentConnectLastUs=0;
        currentConnectName=null;
        synchronized(receiveDataList)
        {
            for(LinkedList<ReceiveDataEvent> receiveDataEvents: receiveDataList)
            {
                receiveDataEvents.clear();
            }
            receiveDataList.notifyAll();
        }
        for(int sendCh=0;sendCh<sendDataList.length;++sendCh)
        {
            synchronized(sendDataList[sendCh])
            {
                sendDataList[sendCh].clear();
            }
        }
    }

    /**
     * 获取配置用的面板
     *
     * @return 配置用的面板
     */
    public JPanel getConfigJPanel()
    {
        return new SmctpConfigJPanel(this);
    }

    public int getDesignatedDataChannel()
    {
        return designatedDataChannel;
    }

    public void setDesignatedDataChannel(int designatedDataChannel)
    {
        this.designatedDataChannel=designatedDataChannel;
        CCMConfigManager.setInteger("ccm/communicate/smctp/Smctp/designatedDataChannel",designatedDataChannel);
    }

    public long getInSpeed()
    {
        synchronized(inSpeedHistory)
        {
            inSpeedSum=updateHistory(inSpeedHistory,inSpeedSum,0);
            return (inSpeedHistory.size()>2)?(long)((inSpeedSum-inSpeedHistory.getFirst().getValue())/((inSpeedHistory.getLast().getKey()-inSpeedHistory.getFirst().getKey()+1e-7)*1e-3)):0;
        }
    }

    private long updateHistory(LinkedList<Pair<Long,Long>> history,long historySum,long len)
    {
        long ms=System.currentTimeMillis();
        while(history.size()>0)
        {
            if(history.getFirst().getKey()+1000<ms)
            {
                historySum-=history.getFirst().getValue();
                history.removeFirst();
            }
            else
            {
                break;
            }
        }
        if(len!=0)
        {
            historySum+=len;
            history.add(new Pair<>(ms,len));
        }
        return historySum;
    }

    public int getWaitLength(int id)
    {
        return receiveDataList[id].size();
    }

    public long getOutSpeed()
    {
        synchronized(outSpeedHistory)
        {
            outSpeedSum=updateHistory(outSpeedHistory,outSpeedSum,0);
            return (outSpeedHistory.size()>2)?(long)((outSpeedSum-outSpeedHistory.getFirst().getValue())/((outSpeedHistory.getLast().getKey()-outSpeedHistory.getFirst().getKey()+1e-7)*1e-3)):0;
        }
    }

    /**
     * 接收到数据事件
     */
    @Override
    public void receiveData(ReceiveDataEvent event)
    {
        if(!start)
        {
            return;
        }
        if(event.getSource()==next)
        {
            synchronized(inSpeedHistory)
            {
                inSpeedSum=updateHistory(inSpeedHistory,inSpeedSum,event.getData().length);
                totalBytes+=event.getData().length;
                if(designatedDataChannel!=CHANNEL_MAX)
                {
                    callReceiveDataListener(designatedDataChannel,event);
                    successBytes+=event.getData().length;
                }
                else
                {
                    for(int eventDataI=0;eventDataI<event.getData().length;)
                    {
                        if(RVI.judgeHead(event.getData()[eventDataI]))
                        {
                            stop();
                            System.out.println("smctp newwwwwwww");
                            start();
                            totalBytes+=event.getData().length-eventDataI;  //补偿清零
                        }
                        switch(RVI.state)
                        {
                            case RECEIVING_HEAD ->
                            {
                                RVI.buf[RVI.bufStart]=event.getData()[eventDataI];
                                RVI.bufStart=(RVI.bufStart+1)%RVI.buf.length;
                                ++eventDataI;
                                final int type=Buffer.getUint8(RVI.buf,RVI.bufStart-1);
                                if(Buffer.getUint8(RVI.buf,RVI.bufStart-3)==0XFF&&Buffer.getUint8(RVI.buf,RVI.bufStart-2)==0x12&&type>=0X17&&type<=0X1A)
                                {
                                    final int version=Buffer.getUint8(RVI.buf,RVI.bufStart-(3+2+1));
                                    final boolean absoluteTime=(version&0X40)!=0;
                                    final boolean shortTime=(version&0X80)!=0;
                                    final int headLen=(3+2+1+1+2+(shortTime?4:8));
                                    if(Buffer.getUint16(RVI.buf,RVI.bufStart-5)==Crc.crc16Modbus(RVI.buf,RVI.bufStart-(headLen),headLen-5))
                                    {
                                        final long nowReceiveUs=shortTime?Buffer.getUint32(RVI.buf,RVI.bufStart-(3+2+1+1+2+4)):Buffer.getUint64(RVI.buf,RVI.bufStart-(3+2+1+1+2+8));
                                        RVI.channel=Buffer.getUint8(RVI.buf,RVI.bufStart-(3+2+1+1));
                                        RVI.dataLength=Buffer.getUint16(RVI.buf,RVI.bufStart-(3+2+1+1+2));
                                        RVI.receivedLength=0;
                                        if(type==0X17)
                                        {
                                            if(RVI.channel<CHANNEL_MAX&&RVI.dataLength!=0)
                                            {
                                                RVI.state=ReceiveState.RECEIVING_DATA;
                                                successBytes+=headLen;
                                                lastReceiveUS=nowReceiveUs;
                                                System.out.printf("SMCTP Receive head%s Ch:%2d Si:%6d Ti:%s\n",(shortTime?" ST":""),RVI.channel,RVI.dataLength,timeFormatter(nowReceiveUs));
                                            }
                                            else
                                            {
                                                System.out.printf("SMCTP channel failed%s Ch:%2d Si:%6d Ti:%s\n",(shortTime?" ST":""),RVI.channel,RVI.dataLength,timeFormatter(nowReceiveUs));
                                            }
                                        }
                                        else if(type==0X18)
                                        {
                                            RVI.state=ReceiveState.RECEIVING_0X18;
                                            successBytes+=headLen;
                                            lastReceiveUS=nowReceiveUs;
                                            System.out.printf("SMCTP Receive system info%s Ti:%s\n",(shortTime?" ST":""),timeFormatter(nowReceiveUs));
                                        }
                                        else if(type==0X1A)
                                        {
                                            successBytes+=headLen;
                                            lastReceiveUS=nowReceiveUs;
                                            System.out.printf("SMCTP Receive heart beat%s Ti:%s\n",(shortTime?" ST":""),timeFormatter(nowReceiveUs));
                                        }
                                        else
                                        {
                                            System.out.printf("SMCTP type failed%s %#04X\n",(shortTime?" ST":""),type);
                                        }
                                    }
                                    else
                                    {
                                        System.out.printf("SMCTP CRC failed%s %d\n",(shortTime?" ST":""),headLen);
                                    }
                                }
                            }
                            case RECEIVING_DATA ->
                            {
                                final long len=Math.min(event.getData().length-eventDataI,RVI.dataLength-RVI.receivedLength);
                                successBytes+=len;
                                byte[] data=Arrays.copyOfRange(event.getData(),eventDataI,(int)(eventDataI+len));
                                RVI.receivedLength+=len;
                                eventDataI+=len;
                                System.out.printf("SMCTP Receive Data %d\n",len);
                                synchronized(receiveDataList)
                                {
                                    while(receiveDataList[RVI.channel].size()>=8192&&currentConnectName!=null)
                                    {
                                        try
                                        {
                                            receiveDataList.wait();
                                        }catch(Exception ignore)
                                        {
                                        }
                                    }
                                    receiveDataList[RVI.channel].add(new ReceiveDataEvent(Smctp.this,data,lastReceiveUS));
                                    receiveDataList.notifyAll();
                                }
                                if(RVI.receivedLength>=RVI.dataLength)
                                {
                                    RVI.state=ReceiveState.RECEIVING_HEAD;
                                    System.out.printf("SMCTP Receive  finish    Ch:%2d Si:%6d\n",RVI.channel,RVI.dataLength);
                                }
                            }
                            case RECEIVING_0X18 ->
                            {
                                {
                                    final long len=Math.min(event.getData().length-eventDataI,RVI.dataLength-RVI.receivedLength);
                                    successBytes+=len;
                                    byte[] data=Arrays.copyOfRange(event.getData(),eventDataI,(int)(eventDataI+len));
                                    eventDataI+=len;
                                    for(int i=0;i<len;++i,++RVI.receivedLength)
                                    {
                                        RVI.buf[RVI.receivedLength]=data[i];
                                    }
                                }
                                if(RVI.receivedLength>=RVI.dataLength)
                                {
                                    //id,id,len,name.....name,crc16,crc16
                                    final int len=Buffer.getUint8(RVI.buf,2);
                                    if(len<=240&&Buffer.getUint16(RVI.buf,3+len)==Crc.crc16Modbus(RVI.buf,0,3+len))
                                    {
                                        synchronized(receiveDataList)
                                        {
                                            currentConnectId=Buffer.getUint16(RVI.buf,0);
                                            currentConnectName=new String(Arrays.copyOfRange(RVI.buf,3,3+len));
                                            currentConnectLastUs=lastReceiveUS;
                                            callStatusChangeListener(StatusChangeEvent.SMCTP_GET_NAME);
                                            System.out.printf("SMCTP Receive system info id:%d name:%s\n",currentConnectId,currentConnectName);
                                            receiveDataList.notifyAll();
                                        }
                                    }
                                    else
                                    {
                                        System.out.printf("SMCTP Receive system info crc failed\n");
                                    }
                                    RVI.state=ReceiveState.RECEIVING_HEAD;
                                }
                            }
                        }

                    }
                }
            }
        }
        else
        {
            int callCnt=0;
            for(int i=0;i<receiveDataListener.length;++i)
            {
                if(receiveDataListener[i].contains(event.getSource()))
                {
                    send(event.getData(),i);
                    ++callCnt;
                }
            }
            if(callCnt!=1||event.getSource() instanceof Communicate)
            {
                System.out.println("Communicate warning callCnt="+callCnt);
            }
        }
    }

    /**
     * 启动
     */
    public void start()
    {
        clear();
        next.receiveData(new ReceiveDataEvent(Smctp.this,ReceiveInfo.restartHead));
        next.receiveData(new ReceiveDataEvent(Smctp.this,ReceiveInfo.restartHead));
        next.receiveData(new ReceiveDataEvent(Smctp.this,ReceiveInfo.restartHead));
        next.receiveData(new ReceiveDataEvent(Smctp.this,ReceiveInfo.restartHead));
        start=true;
        callStatusChangeListener(StatusChangeEvent.START);
    }

    /**
     * 停止
     */
    public void stop()
    {
        clear();
        callStatusChangeListener(StatusChangeEvent.STOP);
        start=false;
        clear();
        System.gc();
    }

    /**
     * 检查是否开启
     *
     * @return true表示开启
     * false表示关闭
     */
    public boolean isOpen()
    {
        return start;
    }

    /**
     * 发送数据
     *
     * @param data   数据
     * @param sendCh 通道
     * @throws NotStartException   未启动时抛出此异常
     * @throws SendFailedException 发送失败时抛出此异常
     * @implNote 通道未使用
     */
    private void send(byte[] data,int sendCh) throws NotStartException, SendFailedException
    {
        synchronized(sendDataList[sendCh])
        {
            sendDataList[sendCh].add(data);
        }
// send(data,sendCh,0X17);
    }

    private class Sender implements Runnable
    {
        @Override
        public void run()
        {
            try
            {
                for(int sendCh=0;sendCh<sendDataList.length;++sendCh)
                {
                    synchronized(sendDataList[sendCh])
                    {
                        final int n=sendDataList[sendCh].size();
                        if(n>0)
                        {
                            int size=0;
                            for(int i=0;i<n;++i) size+=sendDataList[sendCh].get(i).length;
                            if(size>0)
                            {
                                byte[] tmp=new byte[size];
                                for(int i=0, j=0;i<n;++i)
                                {
                                    byte[] a=sendDataList[sendCh].get(i);
                                    System.arraycopy(a,0,tmp,j,a.length);
                                    j+=a.length;
                                }
                                send(tmp,sendCh,0X17);
                                sendDataList[sendCh].clear();
                            }
                        }
                    }
                }
            }catch(Exception ex)
            {
            }
        }
    }

    private void send(byte[] data,int sendCh,int type) throws NotStartException, SendFailedException
    {
        if(!isOpen())
        {
            throw new NotStartException();
        }
        synchronized(outSpeedHistory)
        {
            if(designatedDataChannel==CHANNEL_MAX)
            {
                long nowTimeUs=(System.currentTimeMillis()+zoneOffset)*1000;
                final boolean shortTime=nowTimeUs<0XFFFFFFFFL;
                final boolean absoluteTime=true;
                final int version=(smctpVersion&0X3F)|(((absoluteTime?1:0)<<6)&0X40)|(((shortTime?1:0)<<7)&0X80);
                int start=0;
                final int headLen=(3+2+1+1+2+(shortTime?4:8));
                byte[] buf=new byte[headLen];
                start=shortTime?Buffer.setUint32(buf,start,(int)nowTimeUs):Buffer.setUint64(buf,start,nowTimeUs);            //写入时间戳
                start=Buffer.setUint16(buf,start,data.length);          //写入长度
                start=Buffer.setUint8(buf,start,sendCh);                //写入通道
                start=Buffer.setUint8(buf,start,version);               //写入版本
                start=Buffer.setUint16(buf,start,Crc.crc16Modbus(buf,0,start));
                start=Buffer.setUint8(buf,start,0XFF);
                start=Buffer.setUint8(buf,start,0X12);
                start=Buffer.setUint8(buf,start,type);
                synchronized(outSpeedHistory)
                {
                    outSpeedSum=updateHistory(outSpeedHistory,outSpeedSum,buf.length);
                }
                try
                {
                    next.receiveData(new ReceiveDataEvent(Smctp.this,buf));
                }catch(NotStartException e)
                {
                    e.printStackTrace();
                }
            }
            synchronized(outSpeedHistory)
            {
                outSpeedSum=updateHistory(outSpeedHistory,outSpeedSum,data.length);
            }
            try
            {
                next.receiveData(new ReceiveDataEvent(Smctp.this,data));
            }catch(NotStartException e)
            {
                e.printStackTrace();
            }
            sentFlag=true;
        }
    }

    /**
     * 状态变化事件
     *
     * @param event
     */
    public void statusChange(StatusChangeEvent event)
    {
        if(event.getReason()==StatusChangeEvent.START)
        {
            start();
        }
        else if(event.getReason()==StatusChangeEvent.STOP)
        {
            stop();
        }
        else
        {
            callStatusChangeListener(event.getReason());
        }
    }

    /**
     * 获取上一包的时间
     *
     * @return 上一包的时间
     */
    public long getLastReceiveUS()
    {
        return lastReceiveUS;
    }

    protected enum ReceiveState
    {
        RECEIVING_HEAD,RECEIVING_DATA,RECEIVING_0X18
    }

    private static class ReceiveInfo
    {
        private static final byte[] restartHead=new byte[]{((byte)0XCF),((byte)0XD2),((byte)0X02),((byte)0XC3),((byte)0X91),((byte)0XC9),((byte)0X7F),((byte)0XD0),((byte)0X1A),((byte)0X31),((byte)0X8A),((byte)0X8D),((byte)0X5C),((byte)0XC5),0XF,((byte)0XD5),((byte)0XF4),((byte)0XE5),((byte)0X1A),0X2,((byte)0X1A),((byte)0X17),((byte)0XE2),((byte)0XBE),((byte)0X99),((byte)0X38),((byte)0X5F),((byte)0XDA),((byte)0X8C),((byte)0X91),0X7,((byte)0XDF),((byte)0X91),((byte)0XA3),((byte)0XF1),((byte)0X67),((byte)0X88),((byte)0XEB),((byte)0XA8),((byte)0X86),((byte)0X83),((byte)0XA5),((byte)0X48),((byte)0X86),((byte)0X92),((byte)0XDD),((byte)0XCB),((byte)0X4C),((byte)0XD1),((byte)0X4F),((byte)0X52),((byte)0XF5),((byte)0XC7),((byte)0X14),((byte)0XFF),((byte)0X8A),((byte)0X93),((byte)0X61),((byte)0XD2),((byte)0X6F),((byte)0X13),((byte)0X1C),((byte)0X50),((byte)0XE3),};
        protected final byte[] headBuf;
        protected final byte[] buf;
        protected int headStart;
        protected ReceiveState state;
        protected int bufStart;
        protected int channel;
        protected long dataLength;
        protected int receivedLength;
        protected int totalInfoCnt;

        public ReceiveInfo()
        {
            buf=new byte[255];
            headBuf=new byte[restartHead.length];
            clear();
        }

        public void clear()
        {
            state=ReceiveState.RECEIVING_HEAD;
            Arrays.fill(buf,(byte)0);
            Arrays.fill(headBuf,(byte)0);
            headStart=0;
            bufStart=0;
            channel=CHANNEL_MAX;
            dataLength=0;
            receivedLength=0;
        }

        public boolean judgeHead(byte putAndJudge)
        {
            headBuf[headStart]=putAndJudge;
            headStart=(headStart+1)%headBuf.length;
            ++totalInfoCnt;
            for(int i=0;i<restartHead.length;++i)
            {
                if(restartHead[i]!=headBuf[(headStart+i)%headBuf.length])
                {
                    return false;
                }
            }
            if(totalInfoCnt>headBuf.length)
            {
                totalInfoCnt=0;
                return true;
            }
            totalInfoCnt=0;
            return false;
        }
    }

    private class Consumer implements Runnable
    {
        private final int id;

        private Consumer(int id)
        {
            this.id=id;
        }

        @Override
        public void run()
        {
            while(true)
            {
                ReceiveDataEvent pair=null;
                synchronized(receiveDataList)
                {
                    while(currentConnectName==null||receiveDataList[id].isEmpty())
                    {
                        try
                        {
                            receiveDataList.wait();
                        }catch(Exception ignore)
                        {
                        }
                    }
                    pair=receiveDataList[id].pop();
                    if(receiveDataList[id].size()<=1024)
                    {
                        receiveDataList.notifyAll();
                    }
                }
                if(pair!=null)
                {
                    callReceiveDataListener(id,pair);
                }
            }
        }
    }

}
